package com.haima.sage.bigdata.api.common;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.haima.sage.bigdata.api.entity.storage.*;
import io.lettuce.core.RedisURI;
import io.lettuce.core.cluster.RedisClusterClient;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class RedisUtils {
    private static final Logger LOG = LogManager.getLogger(EsUtils.class);
    private static final Map<String, RedisClusterClient> map = new HashMap<>();

    public static synchronized RedisClusterClient getClient(String store) {
        return map.get(store);
    }

    public static void merge(Storage storage) throws IOException {
        RedisClusterClient client = map.get(storage.getId());
        if (client != null) {
            // TODO 清理数据源
            Thread thread = new Thread(() -> {
                try {
                    TimeUnit.SECONDS.sleep(30);
                } catch (InterruptedException e) {

                } finally {
                    client.shutdown();
                }
            });
            thread.setDaemon(true);
            thread.start();
        }
        map.put(storage.getId(), create(storage));
    }

    private static RedisClusterClient create(Storage storage) throws IOException {
        ObjectMapper mapper = new ObjectMapper();
        RedisStorageData config = mapper.readValue(storage.getData(), RedisStorageData.class);
        return newClient(config);
    }

    private static RedisClusterClient newClient(RedisStorageData config) {
        if (config instanceof RedisClusterStorageData) {
            return newClusterClient(((RedisClusterStorageData) config).getClusters());
        } else if (config instanceof RedisSentinelStorageData) {
            return newSentinelClient(((RedisSentinelStorageData) config).getSentinel());
        }
        return newDefaultClient(((RedisDefaultStorageData) config).getRedis());
    }

    public static RedisClusterClient newClusterClient(List<String> clusters) {
        List<RedisURI> list = new ArrayList<>();
        for (String uri : clusters) {
            try {
                list.add(RedisURI.create(uri));
            } catch (Exception e) {
                LOG.error("redis config error:{}", uri, e);
            }
        }
        RedisClusterClient client = RedisClusterClient.create(list);
        return client;
    }

    public static RedisClusterClient newDefaultClient(String redis) {
        RedisClusterClient client = RedisClusterClient.create(redis);
        return client;
    }

    public static RedisClusterClient newSentinelClient(String sentinel) {
        RedisClusterClient client = RedisClusterClient.create(sentinel);
        return client;
    }
}
